/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/*
 * MPSHadoopJob.java
 *
 * Created on Jul 9, 2010, 4:27:13 PM
 */
package homomorphism.list;

import homomorphism.GroupLongPairComparator;
import homomorphism.HomomorphismBase;
import homomorphism.ListPartitioner;
import homomorphism.SecondPartitioner;
import homomorphism.TotalLongPairKeyComparator;
import utility.LongPair;
import java.io.IOException;


import java.util.ArrayList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


import java.lang.reflect.*;
import java.util.Date;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.NullWritable;

/**
 *
 * @author takeshi
 */
//ThirdHomomorphismWrapper
public class ThirdHomomorphismWrapper<InVal, OtVal> extends HomomorphismBase {

    private static final String MyUsage = "usage:  [input file path]  [output file path] <-M [:num]> <-R [:num]>-h  :[usage]";
    private static final Log LOG = LogFactory.getLog(ThirdHomomorphismWrapper.class);
    //  public static final int Giga = 1073741824;
    //  public static final int Mega = 1048576;
    //  public static final int Kilo = 1024;
    //  private static final String UserClassNameProperty = "UserAppName";
    //  private static final String userAppProperty = "UserAppName";
//    private Class FoldingAguClass;
//    private Class UnFoldingAguClass;
//    private String UserClassName;

    public ThirdHomomorphismWrapper() {
    }

    public OtVal fold(ArrayList<InVal> values) throws Exception {
        throw new RuntimeException("Compiled Code");
    }

    public ArrayList<InVal> unfold(OtVal value) throws Exception {
        throw new RuntimeException("Compiled Code");
    }

    public static class MyFirstMapper<InVal, OtVal> extends FirstMapper<InVal, OtVal> {
    }

    public static class MyFirstReducer<InVal, OtVal> extends FirstReducerFor3rdHomo<InVal, OtVal> {
    }

    public static class MySecondMapper<OtVal, InVal> extends SecondMapperFor3rdHomo<OtVal, InVal> {
    }

    public static class MySecondReducer<InVal, OtVal> extends SecondReducerFor3rdHomo<InVal, OtVal> {
    }

    public int run(String[] args) throws Exception {
        if (args.length < 3) {
            printUsage(this, MyUsage);
            return -1;
        }
        Configuration conf = getConf();

        JobConf job1 = new JobConf(conf, this.getClass());
        job1.setJarByClass(this.getClass());

        job1 = initJobConf1(job1, MyFirstMapper.class, MyFirstReducer.class, args);
        if (job1 == null) {
            //        LOG.error(" job1 ini  failed ");
            return -1;
        }
        JobConf job2 = new JobConf(getConf(), this.getClass());
        job2.setJarByClass(this.getClass());
        job2 = initJobConf2(job2, MySecondMapper.class, MySecondReducer.class, args);
        // LOG.info("job generating finished!");

        Date startTime = new Date();
        System.out.println("My Job started: " + startTime);


        if (JobClient.runJob(job1).isSuccessful()) {
            JobClient.runJob(job2);
            Date end_time = new Date();
            System.out.println("My Job ended: " + end_time);
            System.out.println("The 2 job took "
                    + (end_time.getTime() - startTime.getTime()) / 1000 + " seconds.");
            return 0;
        } else {
            //         LOG.error(" job1 excuting  failed ");
            return -1;
        }
    }
    public String inputPaths;

    public void setInputPaths(String inputPaths) {
        this.inputPaths = inputPaths;
    }

    public String getInputPaths() {
        return inputPaths;
    }
    public String outputPath;

    public void setOutputPath(String outputPath) {
        this.outputPath = outputPath;
    }

    public String getOutputPath() {
        return outputPath;
    }
    public RunningJob runningJob;

    public RunningJob getRunningJob() {
        return runningJob;
    }

    public static void printUsage(Tool tool, String extraArgsUsage) {
        System.err.printf("Usage: %s [genericOptions] %s\n\n",
                tool.getClass().getSimpleName(), extraArgsUsage);
        GenericOptionsParser.printGenericCommandUsage(System.err);
    }

    public JobConf initJobConf1(JobConf job,
            // Class UserJarClass,
            Class UserFirstMapperClass,
            Class UserFirstReducerClass,
            String[] args) throws Exception {

        SequenceFileInputFormat.addInputPath(job, new Path(args[0]));
        SequenceFileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setJobName("ListHomoMorphismStep1");
        job.setMapperClass(UserFirstMapperClass);
        job.setReducerClass(UserFirstReducerClass);

        // job.setOutputValueClass(LongPair.class); UnFoldingAguClass
        ClassLoader myloader = ThirdHomomorphismWrapper.class.getClassLoader();
        String KeyClassName = job.get(IndexKeyClassProperty);
        Class KeyClass = null;
        if (KeyClassName != null && KeyClassName.length() > 0) {
            KeyClass = myloader.loadClass(KeyClassName);
            job.setOutputKeyClass(KeyClass);
            job.setMapOutputKeyClass(KeyClass);
        } else {//default
            job.setOutputKeyClass(LongPair.class); //default LongPair
            job.setMapOutputKeyClass(LongPair.class);
        }
        String OutputClassName = job.get(FoldingOutValClassProperty);
        job.setOutputValueClass(myloader.loadClass(OutputClassName));

        String MapOutputClassName = job.get(FoldingInValClassProperty);
        job.setMapOutputValueClass(myloader.loadClass(MapOutputClassName));
        job.setOutputKeyComparatorClass(TotalLongPairKeyComparator.class);
        job.setOutputValueGroupingComparator(GroupLongPairComparator.class);
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setOutputFormat(SequenceFileOutputFormat.class);
        job.setPartitionerClass(ListPartitioner.class);
//        int mapNum = 1;
//        int idx = HasParameter("-M", args);
//        if (idx >= 0 && args.length > idx) {
//            mapNum = Integer.parseInt(args[idx + 1]);
//        }
//
//        job.setNumMapTasks(mapNum);
        int redNum = 1;
        int idx = HasParameter("-R", args);
        if (idx >= 0 && args.length > idx) {
            redNum = Integer.parseInt(args[idx + 1]);
        }
        job.setNumReduceTasks(redNum);
        //job.setCombinerClass(UserFirstReducerClass);
        return job;
    }

    public JobConf initJobConf2(JobConf job,
            Class User2ndMapperClass,
            Class User2ndReducerClass,
            String[] args)
            throws Exception {
        
        SequenceFileInputFormat.addInputPath(job, new Path(args[1]));
       

        TextOutputFormat.setOutputPath(job, new Path(args[1] + "_final"));

        job.setJobName("ListHomoMorphismStep2");
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setOutputFormat(TextOutputFormat.class);
        job.setMapperClass(User2ndMapperClass);

        ClassLoader myloader = ThirdHomomorphismWrapper.class.getClassLoader();
        String KeyClassName = job.get(IndexKeyClassProperty);
        Class KeyClass = null;
        if (KeyClassName != null && KeyClassName.length() > 0) {
            KeyClass = myloader.loadClass(KeyClassName);
            // job.setOutputKeyClass(KeyClass);
            job.setMapOutputKeyClass(KeyClass);
        } else {//default
          //  job.setOutputKeyClass(LongPair.class); //default LongPair
            job.setMapOutputKeyClass(LongPair.class);
        }
        String OutputClassName = job.get(FoldingOutValClassProperty);
        job.setOutputValueClass(myloader.loadClass(OutputClassName));

        String MapOutputClassName = job.get(FoldingInValClassProperty);
        job.setMapOutputValueClass(myloader.loadClass(MapOutputClassName));
        job.setReducerClass(User2ndReducerClass);
        
        job.setNumReduceTasks(1);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputKeyComparatorClass(TotalLongPairKeyComparator.class);
        job.setOutputValueGroupingComparator(GroupLongPairComparator.class);
        job.setPartitionerClass(SecondPartitioner.class);
        //job.setCombinerClass(FirstReducer.class);
        
        return job;
    }

    /**
     * The main method.
     *
     * usage: -in :[input file path] -out :[output file path] -h  :[usage]
     *
     */
    public static void main(String[] args) throws Exception {

        if (args.length < 3) {
            //   System.out.println("the args.length = " + args.length);
            for (int i = 0; i < args.length; i++) {
                System.err.println(args[i]);
            }
            System.err.println("Usage:hadoop jar  <input path> <output path> <userClassConfigile> ");
            System.exit(-1);
        }

        Configuration conf = new Configuration();
        // add user conf file
        Path userConfFile = new Path(args[2]);
        conf.addResource(userConfFile);
        JobConf job = new JobConf(conf);
        int res = ToolRunner.run(job, new ThirdHomomorphismWrapper(), args);
        System.exit(res);

    }

    public static int HasParameter(String query, String[] pars) {
        boolean rst = false;
        int id = -1; //pars.length;
        for (String par : pars) {
            id++;
            if (par.equals(query)) {
                rst = true;
                //  id ++;
                break;
            }
        }
        if (!rst) {
            id = -1;
        }
        return id;
    }

//    public void setFoldUnfoldArgumentsType(Class<?> userclass, JobConf job) throws Exception {
//
//        ParameterizedType aguType = (ParameterizedType) userclass.getGenericInterfaces()[0];
//        if (aguType.getActualTypeArguments().length != 2) {
//            //  throw new Exception(this.getClass().getSimpleName() + "does not implement right interface!\n");
//        }
//        Class lFoldingAguClass = aguType.getActualTypeArguments()[0].getClass();
//        Class lUnFoldingAguClass = aguType.getActualTypeArguments()[1].getClass();
//        job.setMapOutputValueClass(lUnFoldingAguClass);
//        job.setOutputValueClass(lFoldingAguClass);
//    }
}
